package com.dai.rpc.serialize;


import com.dai.rpc.constant.SerializationTypeEnum;
import com.dai.rpc.exceptions.SerializeException;
import com.dai.rpc.message.RpcRequest;
import com.dai.rpc.message.RpcResponse;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

public class KryoSerializer implements Serializer {


    @Override
    public String name() {
        return SerializationTypeEnum.KRYO.getName();
    }

    /**
     * 由于 Kryo 不是线程安全的，每个线程都应该有自己的 Kryo, Input 和 Output 实例。所以使用 ThreadLocal 存放 Kryo 对象
     * @param obj 要序列化的对象
     * @return
     */

    private static final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(()->{
        Kryo kryo = new Kryo();
        kryo.register(RpcResponse.class);
        kryo.register(RpcRequest.class);
        // 关闭默认注册行为
        kryo.setReferences(true);
        // 关闭循环引用
        kryo.setRegistrationRequired(false);
        return kryo;
    });


    /**
     * 序列化
     * @param obj 要序列化的对象
     * @return
     */
    @Override
    public byte[] serialize(Object obj) {

        try {
            ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
            Output output = new Output(byteArrayOutputStream);
            Kryo kryo = kryoThreadLocal.get();
            kryo.writeObject(output, obj);
            kryoThreadLocal.remove();
            return output.toBytes();
        }catch (Exception e){
            throw new SerializeException("序列化失败！");
        }

    }


    @Override
    public Object deserialize(byte[] bytes, Class<?> clazz) {

        try {
            ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
            Input input = new Input(byteArrayInputStream);
            Kryo kryo = kryoThreadLocal.get();
            Object object = kryo.readObject(input, clazz);
            kryoThreadLocal.remove();
            return object;
        }catch (Exception e){
            e.printStackTrace();
            throw new SerializeException("反序列化失败！");
        }
    }

   /* *//**
     * 反序列化
     * @param bytes 需要反序列化的字节数组
     * @param tClass 目标对象的类型
     * @param <T>
     * @return
     *//*
    @Override
    public <T> T deserialize(byte[] bytes, Class<T> tClass) {

        try {
            ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
            Input input = new Input(byteArrayInputStream);
            Kryo kryo = kryoThreadLocal.get();
            Object object = kryo.readObject(input, tClass);
            kryoThreadLocal.remove();
            return tClass.cast(object);
        }catch (Exception e){
            throw new SerializeException("反序列化失败！");
        }

    }*/
}
